/*
 *
 * FileName: ZkEventThread.java
 * Author:   wx:fdzhangc
 * Date:     2021/11/24 10:44
 * Description: //模块目的、功能描述
 * History: //修改记录
 * <author>      <time>      <version>    <desc>
 * 修改人姓名    修改时间    版本号       描述
 */
package com.zhangc.zcscm.zookeeper;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zhangc.zcscm.zookeeper.exception.ZkInterruptedException;

/**
 * 功能描述:<br>
 *
 * @author wx:fdzhangc
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本] （可选）
 */
public class ZkEventThread extends Thread {
    private static final Logger LOG = LoggerFactory.getLogger(ZkEventThread.class);
    private BlockingQueue<ZkEvent> _events = new LinkedBlockingQueue();
    private static AtomicInteger _eventId = new AtomicInteger(0);

    ZkEventThread(String name) {
        this.setDaemon(true);
        this.setName("ZkClient-EventThread-" + this.getId() + "-" + name);
    }

    public void run() {
        LOG.info("Starting ZkClient event thread.");

        int eventId;
        try {
            for (; !this.isInterrupted(); LOG.debug("Delivering event #" + eventId + " done")) {
                ZkEventThread.ZkEvent zkEvent = (ZkEventThread.ZkEvent) this._events.take();
                eventId = _eventId.incrementAndGet();
                LOG.debug("Delivering event #" + eventId + " " + zkEvent);

                try {
                    zkEvent.run();
                } catch (InterruptedException var4) {
                    this.interrupt();
                } catch (ZkInterruptedException var5) {
                    this.interrupt();
                } catch (Throwable var6) {
                    LOG.error("Error handling event " + zkEvent, var6);
                }
            }
        } catch (InterruptedException var7) {
            LOG.info("Terminate ZkClient event thread.");
        }

    }

    public void send(ZkEventThread.ZkEvent event) {
        if (!this.isInterrupted()) {
            LOG.debug("New event: " + event);
            this._events.add(event);
        }

    }

    abstract static class ZkEvent {
        private String _description;

        public ZkEvent(String description) {
            this._description = description;
        }

        public abstract void run() throws Exception;

        public String toString() {
            return "ZkEvent[" + this._description + "]";
        }
    }
}
